package com.briup.prs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;


/**
 * @author adam
 * @date 2022/2/8
 * 计算用户对于商品的偏好值
 */
public class Step1 extends Configured implements Tool {

    /**
     * 偏好=购买（50%） 订单（30%），购物车（15%），浏览（5%）
     */
    public static class S1Mapper extends Mapper<LongWritable, Text,Text, DoubleWritable> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String[] data = value.toString().split("[,]");
            //将各种行为转换成偏好值
            double like=0;
            if(data.length>=3) {
                String str = data[2].toLowerCase();//有可能数组下标越界
                if (str.equals("addcart")||str.equals("addshopcar")) {
                    like = 0.15;
                } else if (str.equals("showproduct")||str.equals("viewshop")) {
                    like = 0.05;
                } else if (str.equals("createorder")) {
                    like = 0.3;
                } else if(str.equals("paysuccess")) {
                    like = 0.5;
                }
                Text outkey =
                        new Text(data[0] + ":" + data[1]);
                DoubleWritable outvalue =
                        new DoubleWritable(like);
                context.write(outkey, outvalue);
            }
        }
    }
    public static class   S1Reducer extends Reducer<Text,DoubleWritable,Text,DoubleWritable>{
        @Override
        protected void reduce(Text key, Iterable<DoubleWritable> values, Context context) throws IOException, InterruptedException {
            double sum=0.0;
            for (DoubleWritable value : values) {
                sum+=value.get();
            }
            context.write(key,new DoubleWritable(sum));
        }
    }
    @Override
    public int run(String[] strings) throws Exception {
        Configuration conf = getConf();
        String in = conf.get("in");
        String out = conf.get("out");
        Job job = Job.getInstance(conf,"Step1");
        job.setJarByClass(Step1.class);
        job.setInputFormatClass(TextInputFormat.class);
        TextInputFormat.addInputPath(job,new Path(in));
        job.setMapperClass(S1Mapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(DoubleWritable.class);

        job.setReducerClass(S1Reducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(DoubleWritable.class);

        job.setOutputFormatClass(TextOutputFormat.class);
        TextOutputFormat.setOutputPath(job, new Path(out));
        job.waitForCompletion(true);

        return 0;
    }

    public static void main(String[] args) throws Exception {
        ToolRunner.run(new Step1(),args);
    }
}
